package com.sharif.cn.local.subscriber.reactive;

import com.sharif.cn.Subscribe;
import com.sharif.cn.distributed.zookeeper.ZookeeperPublish;
import org.apache.zookeeper.KeeperException;
import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;

import java.io.IOException;

/**
 * 业务数据订阅者
 *
 * @param <T>
 */
public class BusinessSubscriber<T> implements CoreSubscriber<T> {

    private Subscribe subscribe;

    private Subscription subscription;

    private int count = 0;

    private final long maxRequest;

    public BusinessSubscriber(long maxRequest) throws IOException, InterruptedException {
        this.maxRequest = maxRequest;
        subscribe = new Subscribe();
    }

    @Override
    public void onSubscribe(Subscription s) {
        this.subscription = s;
        this.subscription.request(maxRequest);
    }

    @Override
    public void onNext(Object obj) {
        subscribe.onNext(obj);
    }

    @Override
    public void onError(Throwable t) {
        System.out.println("遇到异常：" + t);
    }

    @Override
    public void onComplete() {
        System.out.println("收到数据完成");
    }
}
